/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.hadoop.hbase;

import java.io.IOException;
import java.security.PrivilegedExceptionAction;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

import org.apache.yetus.audience.InterfaceAudience;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.regionserver.HRegionServer;
import org.apache.hadoop.hbase.security.User;
import org.apache.hadoop.hbase.util.JVMClusterUtil.RegionServerThread;
import org.apache.hadoop.hbase.util.Threads;

import java.util.concurrent.CopyOnWriteArrayList;

import org.apache.hadoop.hbase.master.HMaster;
import org.apache.hadoop.hbase.util.JVMClusterUtil;

/**
 * This class creates a single process HBase cluster. One thread is created for
 * a master and one per region server.
 *
 * Call {@link #startup()} to start the cluster running and {@link #shutdown()}
 * to close it all down. {@link #join} the cluster is you want to wait on
 * shutdown completion.
 *
 * <p>Runs master on port 16000 by default.  Because we can't just kill the
 * process -- not till HADOOP-1700 gets fixed and even then.... -- we need to
 * be able to find the master with a remote client to run shutdown.  To use a
 * port other than 16000, set the hbase.master to a value of 'local:PORT':
 * that is 'local', not 'localhost', and the port number the master should use
 * instead of 16000.
 *
 */
@InterfaceAudience.Public
public class LocalHBaseCluster {
    private static final Logger LOG = LoggerFactory.getLogger(LocalHBaseCluster.class);
    private final List<JVMClusterUtil.MasterThread> masterThreads = new CopyOnWriteArrayList<>();
    private final List<JVMClusterUtil.RegionServerThread> regionThreads = new CopyOnWriteArrayList<>();
    private final static int DEFAULT_NO = 1;
    /** local mode */
    public static final String LOCAL = "local";
    /** 'local:' */
    public static final String LOCAL_COLON = LOCAL + ":";
    public static final String ASSIGN_RANDOM_PORTS = "hbase.localcluster.assign.random.ports";

    private final Configuration conf;
    private final Class<? extends HMaster> masterClass;
    private final Class<? extends HRegionServer> regionServerClass;

    /**
     * Constructor.
     * @param conf
     * @throws IOException
     */
    public LocalHBaseCluster(final Configuration conf) throws IOException {
        this(conf, DEFAULT_NO);
    }

    /**
     * Constructor.
     * @param conf Configuration to use.  Post construction has the master's
     * address.
     * @param noRegionServers Count of regionservers to start.
     * @throws IOException
     */
    public LocalHBaseCluster(final Configuration conf, final int noRegionServers) throws IOException {
        this(conf, 1, noRegionServers, getMasterImplementation(conf), getRegionServerImplementation(conf));
    }

    /**
     * Constructor.
     * @param conf Configuration to use.  Post construction has the active master
     * address.
     * @param noMasters Count of masters to start.
     * @param noRegionServers Count of regionservers to start.
     * @throws IOException
     */
    public LocalHBaseCluster(final Configuration conf, final int noMasters, final int noRegionServers) throws IOException {
        this(conf, noMasters, noRegionServers, getMasterImplementation(conf), getRegionServerImplementation(conf));
    }

    @SuppressWarnings("unchecked")
    private static Class<? extends HRegionServer> getRegionServerImplementation(final Configuration conf) {
        return (Class<? extends HRegionServer>) conf.getClass(HConstants.REGION_SERVER_IMPL, HRegionServer.class);
    }

    @SuppressWarnings("unchecked")
    private static Class<? extends HMaster> getMasterImplementation(final Configuration conf) {
        return (Class<? extends HMaster>) conf.getClass(HConstants.MASTER_IMPL, HMaster.class);
    }

    /**
     * Constructor.
     * @param conf Configuration to use.  Post construction has the master's
     * address.
     * @param noMasters Count of masters to start.
     * @param noRegionServers Count of regionservers to start.
     * @param masterClass
     * @param regionServerClass
     * @throws IOException
     */
    @SuppressWarnings("unchecked")
    public LocalHBaseCluster(final Configuration conf, final int noMasters, final int noRegionServers, final Class<? extends HMaster> masterClass,
            final Class<? extends HRegionServer> regionServerClass) throws IOException {
        this.conf = conf;

        // When active, if a port selection is default then we switch to random
        if(conf.getBoolean(ASSIGN_RANDOM_PORTS, false)) {
            if(conf.getInt(HConstants.MASTER_PORT, HConstants.DEFAULT_MASTER_PORT) == HConstants.DEFAULT_MASTER_PORT) {
                LOG.debug("Setting Master Port to random.");
                conf.set(HConstants.MASTER_PORT, "0");
            }
            if(conf.getInt(HConstants.REGIONSERVER_PORT, HConstants.DEFAULT_REGIONSERVER_PORT) == HConstants.DEFAULT_REGIONSERVER_PORT) {
                LOG.debug("Setting RegionServer Port to random.");
                conf.set(HConstants.REGIONSERVER_PORT, "0");
            }
            // treat info ports special; expressly don't change '-1' (keep off)
            // in case we make that the default behavior.
            if(conf.getInt(HConstants.REGIONSERVER_INFO_PORT, 0) != -1 && conf.getInt(HConstants.REGIONSERVER_INFO_PORT,
                    HConstants.DEFAULT_REGIONSERVER_INFOPORT) == HConstants.DEFAULT_REGIONSERVER_INFOPORT) {
                LOG.debug("Setting RS InfoServer Port to random.");
                conf.set(HConstants.REGIONSERVER_INFO_PORT, "0");
            }
            if(conf.getInt(HConstants.MASTER_INFO_PORT, 0) != -1 && conf
                    .getInt(HConstants.MASTER_INFO_PORT, HConstants.DEFAULT_MASTER_INFOPORT) == HConstants.DEFAULT_MASTER_INFOPORT) {
                LOG.debug("Setting Master InfoServer Port to random.");
                conf.set(HConstants.MASTER_INFO_PORT, "0");
            }
        }

        this.masterClass = (Class<? extends HMaster>) conf.getClass(HConstants.MASTER_IMPL, masterClass);
        // Start the HMasters.
        for(int i = 0; i < noMasters; i++) {
            addMaster(new Configuration(conf), i);
        }
        // Start the HRegionServers.
        this.regionServerClass = (Class<? extends HRegionServer>) conf.getClass(HConstants.REGION_SERVER_IMPL, regionServerClass);

        for(int i = 0; i < noRegionServers; i++) {
            addRegionServer(new Configuration(conf), i);
        }
    }

    public JVMClusterUtil.RegionServerThread addRegionServer() throws IOException {
        return addRegionServer(new Configuration(conf), this.regionThreads.size());
    }

    @SuppressWarnings("unchecked")
    public JVMClusterUtil.RegionServerThread addRegionServer(Configuration config, final int index) throws IOException {
        // Create each regionserver with its own Configuration instance so each has
        // its Connection instance rather than share (see HBASE_INSTANCES down in
        // the guts of ConnectionManager).
        JVMClusterUtil.RegionServerThread rst = JVMClusterUtil.createRegionServerThread(config,
                (Class<? extends HRegionServer>) conf.getClass(HConstants.REGION_SERVER_IMPL, this.regionServerClass), index);

        this.regionThreads.add(rst);
        return rst;
    }

    public JVMClusterUtil.RegionServerThread addRegionServer(final Configuration config, final int index,
            User user) throws IOException, InterruptedException {
        return user.runAs(new PrivilegedExceptionAction<JVMClusterUtil.RegionServerThread>() {
            @Override
            public JVMClusterUtil.RegionServerThread run() throws Exception {
                return addRegionServer(config, index);
            }
        });
    }

    public JVMClusterUtil.MasterThread addMaster() throws IOException {
        return addMaster(new Configuration(conf), this.masterThreads.size());
    }

    public JVMClusterUtil.MasterThread addMaster(Configuration c, final int index) throws IOException {
        // Create each master with its own Configuration instance so each has
        // its Connection instance rather than share (see HBASE_INSTANCES down in
        // the guts of ConnectionManager.
        JVMClusterUtil.MasterThread mt = JVMClusterUtil
                .createMasterThread(c, (Class<? extends HMaster>) conf.getClass(HConstants.MASTER_IMPL, this.masterClass), index);
        this.masterThreads.add(mt);
        return mt;
    }

    public JVMClusterUtil.MasterThread addMaster(final Configuration c, final int index, User user) throws IOException, InterruptedException {
        return user.runAs(new PrivilegedExceptionAction<JVMClusterUtil.MasterThread>() {
            @Override
            public JVMClusterUtil.MasterThread run() throws Exception {
                return addMaster(c, index);
            }
        });
    }

    /**
     * @param serverNumber
     * @return region server
     */
    public HRegionServer getRegionServer(int serverNumber) {
        return regionThreads.get(serverNumber).getRegionServer();
    }

    /**
     * @return Read-only list of region server threads.
     */
    public List<JVMClusterUtil.RegionServerThread> getRegionServers() {
        return Collections.unmodifiableList(this.regionThreads);
    }

    /**
     * @return List of running servers (Some servers may have been killed or
     * aborted during lifetime of cluster; these servers are not included in this
     * list).
     */
    public List<JVMClusterUtil.RegionServerThread> getLiveRegionServers() {
        List<JVMClusterUtil.RegionServerThread> liveServers = new ArrayList<>();
        List<RegionServerThread> list = getRegionServers();
        for(JVMClusterUtil.RegionServerThread rst : list) {
            if(rst.isAlive())
                liveServers.add(rst);
            else
                LOG.info("Not alive " + rst.getName());
        }
        return liveServers;
    }

    /**
     * @return the Configuration used by this LocalHBaseCluster
     */
    public Configuration getConfiguration() {
        return this.conf;
    }

    /**
     * Wait for the specified region server to stop. Removes this thread from list of running threads.
     * @return Name of region server that just went down.
     */
    public String waitOnRegionServer(int serverNumber) {
        JVMClusterUtil.RegionServerThread regionServerThread = this.regionThreads.get(serverNumber);
        return waitOnRegionServer(regionServerThread);
    }

    /**
     * Wait for the specified region server to stop. Removes this thread from list of running threads.
     * @return Name of region server that just went down.
     */
    public String waitOnRegionServer(JVMClusterUtil.RegionServerThread rst) {
        while(rst.isAlive()) {
            try {
                LOG.info("Waiting on " + rst.getRegionServer().toString());
                rst.join();
            } catch(InterruptedException e) {
                e.printStackTrace();
            }
        }
        regionThreads.remove(rst);
        return rst.getName();
    }

    /**
     * @return the HMaster thread
     */
    public HMaster getMaster(int serverNumber) {
        return masterThreads.get(serverNumber).getMaster();
    }

    /**
     * Gets the current active master, if available.  If no active master, returns
     * null.
     * @return the HMaster for the active master
     */
    public HMaster getActiveMaster() {
        for(JVMClusterUtil.MasterThread mt : masterThreads) {
            // Ensure that the current active master is not stopped.
            // We don't want to return a stopping master as an active master.
            if(mt.getMaster().isActiveMaster() && !mt.getMaster().isStopped()) {
                return mt.getMaster();
            }
        }
        return null;
    }

    /**
     * @return Read-only list of master threads.
     */
    public List<JVMClusterUtil.MasterThread> getMasters() {
        return Collections.unmodifiableList(this.masterThreads);
    }

    /**
     * @return List of running master servers (Some servers may have been killed
     * or aborted during lifetime of cluster; these servers are not included in
     * this list).
     */
    public List<JVMClusterUtil.MasterThread> getLiveMasters() {
        List<JVMClusterUtil.MasterThread> liveServers = new ArrayList<>();
        List<JVMClusterUtil.MasterThread> list = getMasters();
        for(JVMClusterUtil.MasterThread mt : list) {
            if(mt.isAlive()) {
                liveServers.add(mt);
            }
        }
        return liveServers;
    }

    /**
     * Wait for the specified master to stop. Removes this thread from list of running threads.
     * @return Name of master that just went down.
     */
    public String waitOnMaster(int serverNumber) {
        JVMClusterUtil.MasterThread masterThread = this.masterThreads.get(serverNumber);
        return waitOnMaster(masterThread);
    }

    /**
     * Wait for the specified master to stop. Removes this thread from list of running threads.
     * @return Name of master that just went down.
     */
    public String waitOnMaster(JVMClusterUtil.MasterThread masterThread) {
        while(masterThread.isAlive()) {
            try {
                LOG.info("Waiting on " + masterThread.getMaster().getServerName().toString());
                masterThread.join();
            } catch(InterruptedException e) {
                e.printStackTrace();
            }
        }
        masterThreads.remove(masterThread);
        return masterThread.getName();
    }

    /**
     * Wait for Mini HBase Cluster to shut down.
     * Presumes you've already called {@link #shutdown()}.
     */
    public void join() {
        if(this.regionThreads != null) {
            for(Thread t : this.regionThreads) {
                if(t.isAlive()) {
                    try {
                        Threads.threadDumpingIsAlive(t);
                    } catch(InterruptedException e) {
                        LOG.debug("Interrupted", e);
                    }
                }
            }
        }
        if(this.masterThreads != null) {
            for(Thread t : this.masterThreads) {
                if(t.isAlive()) {
                    try {
                        Threads.threadDumpingIsAlive(t);
                    } catch(InterruptedException e) {
                        LOG.debug("Interrupted", e);
                    }
                }
            }
        }
    }

    /**
     * Start the cluster.
     */
    public void startup() throws IOException {
        JVMClusterUtil.startup(this.masterThreads, this.regionThreads);
    }

    /**
     * Shut down the mini HBase cluster
     */
    public void shutdown() {
        JVMClusterUtil.shutdown(this.masterThreads, this.regionThreads);
    }

    /**
     * @param c Configuration to check.
     * @return True if a 'local' address in hbase.master value.
     */
    public static boolean isLocal(final Configuration c) {
        boolean mode = c.getBoolean(HConstants.CLUSTER_DISTRIBUTED, HConstants.DEFAULT_CLUSTER_DISTRIBUTED);
        return (mode == HConstants.CLUSTER_IS_LOCAL);
    }

    /**
     * Test things basically work.
     * @param args
     * @throws IOException
     */
    public static void main(String[] args) throws IOException {
        Configuration conf = HBaseConfiguration.create();
        LocalHBaseCluster cluster = new LocalHBaseCluster(conf);
        cluster.startup();
        Connection connection = ConnectionFactory.createConnection(conf);
        Admin admin = connection.getAdmin();
        try {
            HTableDescriptor htd = new HTableDescriptor(TableName.valueOf(cluster.getClass().getName()));
            admin.createTable(htd);
        } finally {
            admin.close();
        }
        connection.close();
        cluster.shutdown();
    }
}
